package com.example.monit.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.sql.QueryResponse;
import co.elastic.clients.json.JsonData;
import com.example.monit.bean.BasicBehavior;
import com.example.monit.bean.ResponseRust;
import com.example.monit.dto.QueryBasicBehaviorDto;
import com.example.monit.dto.TotalDto;
import com.example.monit.enums.IndexName;
import com.example.monit.utils.TimeUtils;
import com.example.monit.vo.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@Slf4j
@Service
public class BasicBehaviorServiceImpl implements BasicBehaviorService {

    @Autowired
    ElasticsearchClient client;

    @Override
    public ResponseRust upLoadBasicBehavior(List<BasicBehavior> basicBehaviors) throws IOException {
//        List<BulkOperation> operationList = basicBehaviors.stream().map(e ->
//                        BulkOperation.of(builder ->
//                                builder.index(builder1 -> builder1.document(e).index(IndexName.BasicBehavior))))
//                .collect(Collectors.toList());
//        BulkResponse response = client.bulk(builder -> builder.operations(operationList));
//        if (response.errors()) {
//            log.error("上报数据异常错误");
//        }
        return ResponseRust.success_creat();
    }

    @Override
    public ResponseRust queryBasicBehavior(BasicBehaviorVo basicBehaviorVo) throws IOException {
        String sql = " SELECT pageUrl,count(pageUrl),userID,sum(value)\n" +
                "            FROM \"basic_behavior\"\n" +
                "            where appId=%s and mainType=%d and subType=%d and startTime between %d and %d\n" +
                "            group by pageUrl ,userID\n" +
                "            order by count(pageUrl) desc";

        sql = String.format(sql,
                basicBehaviorVo.getApp_id(),
                basicBehaviorVo.getMain_type(),
                basicBehaviorVo.getSub_type(),
                basicBehaviorVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                basicBehaviorVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli() //todo 时区问题
        );
        String finalSql = sql;
        QueryResponse query = client.sql().query(e -> e.query(finalSql));

        List<List<JsonData>> list = query.rows();

        HashMap<String, QueryBasicBehaviorDto> map = new HashMap<>();
        for (List<JsonData> jsonData : list) {
            String pageUrl = jsonData.get(0).to(String.class);
            Integer count = jsonData.get(1).to(Integer.class);
            String userID = jsonData.get(2).to(String.class);
            BigDecimal sumValue = jsonData.get(3).to(BigDecimal.class);

            QueryBasicBehaviorDto build = QueryBasicBehaviorDto.builder()
                    .pageUrl(pageUrl)
                    .count(count)
                    .userList(Arrays.asList(userID))
                    .pageList(Arrays.asList(pageUrl))
                    .userCount(1)
                    .pageCount(1)
                    .sumAverage(sumValue)
                    .build();

            String key = pageUrl;
            QueryBasicBehaviorDto queryDto = map.get(key);
            if (ObjectUtils.isEmpty(queryDto)) {
                map.put(pageUrl, build);
            } else {
                if (!queryDto.getUserList().contains(userID)) {
                    queryDto.setUserCount(queryDto.getUserCount() + 1);
                    queryDto.getUserList().add(userID);
                }
                if (!queryDto.getPageList().contains(pageUrl)) {
                    queryDto.setPageCount(queryDto.getPageCount() + 1);
                    queryDto.getPageList().add(pageUrl);
                }
                queryDto.setCount(queryDto.getCount() + build.getCount());
            }
        }

        Collection<QueryBasicBehaviorDto> values = map.values();
        for (QueryBasicBehaviorDto queryDto : values) {
            if (!queryDto.getSumAverage().equals(BigDecimal.ZERO)) {
                queryDto.setAverage(queryDto.getSumAverage().divide(new BigDecimal(queryDto.getCount()), 2, RoundingMode.HALF_UP));
            }
        }

        if (!ObjectUtils.isEmpty(basicBehaviorVo.getSize())) {
            values = values.stream().limit(basicBehaviorVo.getSize()).collect(Collectors.toList());
        }

        return ResponseRust.success(values);
    }

    @Override
    public ResponseRust totalBasicBehavior(BasicBehaviorTotalVo basicIndicatorTotalVo) throws IOException {
        SearchRequest searchRequest = SearchRequest.of(s ->
                s.index(IndexName.BasicBehavior)
                        .query(getQuery(basicIndicatorTotalVo))
                        .aggregations("count", getAggregation(basicIndicatorTotalVo))
                        .size(0));

        SearchResponse<BasicBehavior> response = client.search(searchRequest, BasicBehavior.class);
        Map<String, Aggregate> aggregations = response.aggregations();
        Aggregate count = aggregations.get("count");

        Buckets<HistogramBucket> buckets = count.histogram().buckets();
        List<TotalDto> data = totalData(buckets, basicIndicatorTotalVo);
        return ResponseRust.success(data);
    }

    private Aggregation getAggregation(BasicBehaviorTotalVo basicIndicatorTotalVo) {

        HashMap<String, Aggregation> aggregationHashMap = new HashMap<>(3);
        aggregationHashMap.put("avg", Aggregation.of(avg -> avg.avg(e -> e.field("value"))));
        aggregationHashMap.put("userCount", Aggregation.of(userCount -> userCount.cardinality(e -> e.field("userID"))));
        aggregationHashMap.put("pageCount", Aggregation.of(pageCount -> pageCount.cardinality(e -> e.field("pageUrl"))));

        if (!StringUtils.hasLength(basicIndicatorTotalVo.getGranularity())) {
            basicIndicatorTotalVo.setGranularity("1d");
        }
        int intValue = Integer.parseInt(basicIndicatorTotalVo.getGranularity().split("[smhdMy]")[0]);
        String type = basicIndicatorTotalVo.getGranularity().split(intValue + "")[1];
        ChronoUnit unit = TimeUtils.getType(type);
        return Aggregation.of(agg -> agg.histogram(histogram ->
                        histogram.field("startTime").interval((double) unit.getDuration().toMillis() * intValue))
                .aggregations(aggregationHashMap));

    }

    private Query getQuery(BasicBehaviorTotalVo basicIndicatorTotalVo) {
        return Query.of(query -> query.bool(
                bool -> bool
                        .must(t -> t.term(term -> term.field("appId").value(e -> e.stringValue(basicIndicatorTotalVo.getApp_id()))))
                        .must(t -> t.term(term -> term.field("mainType").value(e -> e.longValue(basicIndicatorTotalVo.getMain_type()))))
                        .must(t -> t.term(term -> term.field("subType").value(e -> e.longValue(basicIndicatorTotalVo.getSub_type()))))
                        .must(e -> e.range(r ->
                                r.gte(JsonData.of(basicIndicatorTotalVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                                        .lte(JsonData.of(basicIndicatorTotalVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()))
                                        .field("startTime")))));
    }


    private List<TotalDto> totalData(Buckets<HistogramBucket> bucket, BasTotalVo basicindicatorsTotalVo) {
        List<HistogramBucket> list = bucket.array();
        List<TotalDto> linkedList = new LinkedList<>();

        // 间隔单位
        int intValue = Integer.parseInt(basicindicatorsTotalVo.getGranularity().split("[smhdMy]")[0]);
        String type = basicindicatorsTotalVo.getGranularity().split(intValue + "")[1];
        ChronoUnit unit = TimeUtils.getType(type);

        LocalDateTime startTime = basicindicatorsTotalVo.getStart_time();
        LocalDateTime endTime = basicindicatorsTotalVo.getEnd_time();

        // 假如没有数据
        if (ObjectUtils.isEmpty(list)) {
            while (startTime.isBefore(endTime)) {
                TotalDto totalDto = TotalDto.builder().dateTime(startTime).build();
                startTime = startTime.plus(intValue, unit);
                linkedList.add(totalDto);
            }
            return linkedList;
        }

        // 结果集的第一天
        HistogramBucket histogramBucket1 = list.get(0);
        LocalDateTime resultStartTime = TimeUtils.timestamToDatetime(new BigDecimal(histogramBucket1.key() + "").longValue());


        // 结果集的最后一天
        HistogramBucket histogramBucket2 = list.get(list.size() - 1);
        LocalDateTime resultEndTime = TimeUtils.timestamToDatetime(new BigDecimal(histogramBucket2.key() + "").longValue());


        // 填充开始的时间
        while (startTime.until(resultStartTime, unit) != 0) {
            TotalDto totalDto = TotalDto.builder().dateTime(startTime).build();
            startTime = startTime.plus(intValue, unit);
            linkedList.add(totalDto);
        }

        // 填充中间
        int index = 0;
        while (resultStartTime.equals(resultEndTime) ||  resultStartTime.isBefore(resultEndTime)) {
            HistogramBucket histogramBucket = list.get(index);
            Map<String, Aggregate> map = histogramBucket.aggregations();
            TotalDto totalDto = TotalDto.builder()
                    .dateTime(resultStartTime)
                    .count(histogramBucket.docCount())
                    .pageCount(map.get("pageCount").cardinality().value())
                    .userCount(map.get("userCount").cardinality().value())
                    .build();
            linkedList.add(totalDto);
            resultStartTime = resultStartTime.plus(intValue, unit);
            index++;
        }

        resultEndTime = resultEndTime.plus(intValue, unit);

        // 填充结束时间
        while (resultEndTime.until(endTime, unit) > 0) {
            TotalDto totalDto = TotalDto.builder().dateTime(resultEndTime).build();
            resultEndTime = resultEndTime.plus(intValue, unit);
            linkedList.add(totalDto);
        }

        return linkedList;
    }


    @Override
    public ResponseRust getUserActionList(BaseQueryVo querysVo) throws ExecutionException, InterruptedException {

        CompletableFuture<List<Map<String, Object>>> clickBehaviorsList = this.clickbehaviorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> pageSkipBehaviorsList = this.pageskipbehaviorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> routingSkipBehaviorsList = this.routingskipbehaviorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> resourceErrorsList = this.resourceerrorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> javascriptErrorsList = this.javascripterrorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> promiseErrorsList = this.promiseerrorsList(querysVo);
        CompletableFuture<List<Map<String, Object>>> interfaceErrorsList = this.interfaceerrorsList(querysVo);

        List<CompletableFuture<List<Map<String, Object>>>> list = Arrays.asList(clickBehaviorsList, pageSkipBehaviorsList, routingSkipBehaviorsList,
                resourceErrorsList, javascriptErrorsList, promiseErrorsList, interfaceErrorsList);
        // 并发执行全部任务
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(list.toArray(new CompletableFuture[0]));
        // 阻塞全部的并发查询,等待全部查询出来再执行下一步操作
        Void unused = completableFuture.get();
        List<Map<String, Object>> list1 = new LinkedList<>();
        for (CompletableFuture<List<Map<String, Object>>> future : list) {
            list1.addAll(future.get());
        }
        return ResponseRust.success(list1);
    }

    /**
     * 接口
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> interfaceerrorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = "SELECT appId,startTime,pageUrl,userID,statusCode,method,duration,url,mainType,subType,eventId\n" +
                    "      FROM \"interface_indicator\"\n" +
                    "      where appId= '%s' and userID = '%s'  and startTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            LinkedList<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("listType", 3);
                    map.put("title", "接口");
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("statusCode", jsonDataList.get(4).to(Integer.class));
                    map.put("method", jsonDataList.get(5).to(String.class));
                    map.put("duration", jsonDataList.get(6).to(Long.class));
                    map.put("url", jsonDataList.get(7).to(String.class));
                    map.put("mainType", jsonDataList.get(8).to(Integer.class));
                    map.put("subType", jsonDataList.get(9).to(Integer.class));
                    map.put("eventId", jsonDataList.get(10).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });

    }

    /**
     * promise错误
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> promiseerrorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = "SELECT appId,errorTime,pageUrl,userID,stack,mainType,subType,eventId\n" +
                    "      FROM \"promise_error\"\n" +
                    "      where appId= '%s' and userID = '%s' and errorTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "promise错误");
                    map.put("listType", 2);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("stack", jsonDataList.get(4).to(String.class));
                    map.put("mainType", jsonDataList.get(5).to(Integer.class));
                    map.put("subType", jsonDataList.get(6).to(Integer.class));
                    map.put("eventId", jsonDataList.get(7).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }

    /**
     * js错误
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> javascripterrorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = "SELECT appId,errorTime,pageUrl,userID,msg ,line,column,stack,url,mainType,subType,eventId\n" +
                    "      FROM \"javascript_error\"\n" +
                    "      where appId= '%s' and userID = '%s' and errorTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "js错误");
                    map.put("listType", 2);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("msg", jsonDataList.get(4).to(String.class));
                    map.put("line", jsonDataList.get(5).to(Long.class));
                    map.put("column", jsonDataList.get(6).to(Long.class));
                    map.put("stack", jsonDataList.get(7).to(String.class));
                    map.put("url", jsonDataList.get(8).to(String.class));
                    map.put("mainType", jsonDataList.get(9).to(Integer.class));
                    map.put("subType", jsonDataList.get(10).to(Integer.class));
                    map.put("eventId", jsonDataList.get(11).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }

    /**
     * 资源错误
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> resourceerrorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = "SELECT appId,errorTime,pageUrl,userID,resourceType ,path,mainType,subType,eventId\n" +
                    "      FROM \"resource_error\"\n" +
                    "      where appId= '%s' and userID = '%s' and errorTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "资源错误");
                    map.put("listType", 2);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("resourceType", jsonDataList.get(4).to(String.class));
                    map.put("path", jsonDataList.get(5).to(String.class));
                    map.put("mainType", jsonDataList.get(6).to(Integer.class));
                    map.put("subType", jsonDataList.get(7).to(Integer.class));
                    map.put("eventId", jsonDataList.get(8).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }

    /**
     * 路由跳转行为
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> routingskipbehaviorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = "SELECT appId,startTime,pageUrl,userID,\"from\" ,\"to\",params,query,mainType,subType,eventId\n" +
                    "      FROM \"routing_skip_behavior\"\n" +
                    "      where appId= '%s' and userID = '%s' and startTime between %d and %d";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "浏览");
                    map.put("listType", 1);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("from", jsonDataList.get(4).to(String.class));
                    map.put("to", jsonDataList.get(5).to(String.class));
                    map.put("params", jsonDataList.get(6).to(String.class));
                    map.put("query", jsonDataList.get(7).to(String.class));
                    map.put("mainType", jsonDataList.get(8).to(Integer.class));
                    map.put("subType", jsonDataList.get(9).to(Integer.class));
                    map.put("eventId", jsonDataList.get(10).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }

    /**
     * 页面跳转行为
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> pageskipbehaviorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = " SELECT appId,startTime,pageUrl,userID,\"from\" ,\"to\",mainType,subType,eventId\n" +
                    "      FROM \"page_skip_behavior\"\n" +
                    "      where appId= '%s' and userID = '%s' and startTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "浏览");
                    map.put("listType", 1);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("from", jsonDataList.get(4).to(String.class));
                    map.put("to", jsonDataList.get(5).to(String.class));
                    map.put("mainType", jsonDataList.get(6).to(Integer.class));
                    map.put("subType", jsonDataList.get(7).to(Integer.class));
                    map.put("eventId", jsonDataList.get(8).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }

    /**
     * 点击行为
     *
     * @param baseQueryVo
     * @returns
     */
    private CompletableFuture<List<Map<String, Object>>> clickbehaviorsList(BaseQueryVo baseQueryVo) {
        return CompletableFuture.supplyAsync(() -> {
            String sql = " SELECT appId,startTime,pageUrl,userID,top,\"left\" ,\"inner\",target,mainType,subType,eventId\n" +
                    "      FROM \"click_behavior\"\n" +
                    "      where appId= '%s' and userID = '%s' and startTime between %d and %d ";

            String format = String.format(sql,
                    baseQueryVo.getApp_id(),
                    baseQueryVo.getUser_id(),
                    baseQueryVo.getStart_time().toInstant(ZoneOffset.of("+0")).toEpochMilli(),
                    baseQueryVo.getEnd_time().toInstant(ZoneOffset.of("+0")).toEpochMilli()
            );
            List<Map<String, Object>> hashMaps = new LinkedList<>();
            try {
                QueryResponse response = client.sql().query(q -> q.query(format));
                List<List<JsonData>> list = response.rows();
                for (List<JsonData> jsonDataList : list) {
                    Map<String, Object> map = new HashMap<>();
                    map.put("title", "点击行为");
                    map.put("listType", 1);
                    map.put("appId", jsonDataList.get(0).to(String.class));
                    map.put("time", jsonDataList.get(1).to(Long.class));
                    map.put("pageUrl", jsonDataList.get(2).to(String.class));
                    map.put("userID", jsonDataList.get(3).to(String.class));
                    map.put("top", jsonDataList.get(4).to(Double.class));
                    map.put("left", jsonDataList.get(5).to(Double.class));
                    map.put("inner", jsonDataList.get(6).to(String.class));
                    map.put("target", jsonDataList.get(7).to(String.class));
                    map.put("mainType", jsonDataList.get(8).to(Integer.class));
                    map.put("subType", jsonDataList.get(9).to(Integer.class));
                    map.put("eventId", jsonDataList.get(10).to(String.class));
                    hashMaps.add(map);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return hashMaps;
        });
    }
}
